package com.fengzan.hadoop.socket.server;

import com.fengzan.hadoop.shellexcute.ShellExcute;

import java.io.File;
import java.net.*;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;
import java.util.*;


public class SocketServer {


    /**
     * 初始化选择器
     */
    private Selector selector;

    /**
     * 初始化服务器网络通道
     */
    private ServerSocketChannel serverSocketChannel;

    /**
     * 端口
     */
    private static final int PORT = 6655;


    /**
     * 构造方法
     */
    public SocketServer() {

        try {

            //获取选择器
            selector = Selector.open();
            //获取服务器网络通道
            serverSocketChannel = ServerSocketChannel.open();
            //网络地址
            InetSocketAddress inetSocketAddress = new InetSocketAddress(PORT);
            //服务器网络通道绑定网络地址
            serverSocketChannel.socket().bind(inetSocketAddress);
            //设置服务器网络通道非阻塞
            serverSocketChannel.configureBlocking(false);
            //将服务器网络通道注册到选择器上，绑定连接请求事件
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }


    /**
     * 监听客户端请求事件
     */
    public void listen() {
        try {
            Set<SocketChannel> clients = new HashSet<>();
            //无限循环
            while (true) {
                //获取请求数
                int count = selector.select();
                //count大于0，则代表有请求进来
                if (count > 0) {
                    //获取请求集
                    Iterator<SelectionKey> selectionKeyIterator = selector.selectedKeys().iterator();
                    //遍历请求集
                    while (selectionKeyIterator.hasNext()) {
                        //得到请求
                        SelectionKey selectionKey = selectionKeyIterator.next();
                        //连接请求
                        if (selectionKey.isAcceptable()) {
                            //获取客户端网络通道
                            SocketChannel socketChannel = serverSocketChannel.accept();
                            //设置客户端网络通道非阻塞
                            socketChannel.configureBlocking(false);
                            //将客户端网络通道注册到选择器上
                            socketChannel.register(selector, SelectionKey.OP_READ);
                            clients.add(socketChannel);
                            System.out.println("客户端连接" + clients.size() + "个");
                            if (clients.size() >= 3) {
                                System.out.println("执行hadoop安装shell脚本");
                                // 检查是否已经执行 ssh相关命令
                                File lockFile = new File("/usr/local/src/start_hadoop_lock.d");
                                if (!lockFile.exists()) {
                                    System.out.println("--------------ssh-copy Start--------------");
                                    ShellExcute.excuteShell("expect -f /usr/local/src/ssh-copy.exp");
                                    System.out.println("--------------ssh-copy end--------------");
                                    // 执行hadoop启动shell脚本
                                    if (isHadooop1()) {
                                        ShellExcute.excuteShell("/usr/local/src/starthadoop.sh");
                                        lockFile.createNewFile();
                                    }
                                } else {
                                    System.out.println("--------------ssh-copy not need--------------");
                                    if (isHadooop1()) {
                                        ShellExcute.excuteShell("/usr/local/src/starthadoop.sh");
                                    }
                                }
                            }
                        }
/*                        //信息读取请求
                        if (selectionKey.isReadable()) {
                            //客户端信息读取
                            readData(selectionKey);
                        }*/
                        //移除请求
                        selectionKeyIterator.remove();
                    }
                } else {
                    System.out.println("等待...");
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static boolean isHadooop1() throws SocketException, UnknownHostException {
        boolean isHadooop1 = false;
        InetAddress hadoop1 = InetAddress.getByName("hadoop1");
        System.out.println(Arrays.toString(hadoop1.getAddress()));
        Enumeration enumeration = NetworkInterface.getNetworkInterfaces();
        while (enumeration.hasMoreElements()) {
            NetworkInterface network = (NetworkInterface) enumeration.nextElement();
            if (network.isVirtual() || !network.isUp()) {
                continue;
            } else {
                Enumeration addresses = network.getInetAddresses();
                while (addresses.hasMoreElements()) {
                    InetAddress address = (InetAddress) addresses.nextElement();
                    if (address != null && (address instanceof Inet4Address)) {
                        System.out.println(Arrays.toString(address.getAddress()));
                        if (hadoop1.getHostAddress().equals(address.getHostAddress())) {
                            isHadooop1 = true;
                        }
                    }
                }
            }
        }
        return isHadooop1;
    }

    public static void main(String[] args) throws SocketException, UnknownHostException {
        isHadooop1();
    }

    /**
     * 客户端信息读取
     *
     * @param selectionKey
     */
    private void readData(SelectionKey selectionKey) {
        //初始化客户端网络通道
        SocketChannel socketChannel = null;
        try {
            //获取客户端网络通道
            socketChannel = (SocketChannel) selectionKey.channel();
            //创建缓冲区
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            //读取客户端网络通道中的数据到缓冲区
            int count = socketChannel.read(byteBuffer);
            //判断缓冲区中是否有数据
            if (count > 0) {
                //将缓冲区的数据转换位字符串
                String message = new String(byteBuffer.array());
                System.out.println(message.trim());
                //将信息群发到其他客户端
                sendInfoToOtClients(message, socketChannel);
            }

        } catch (Exception e) {

            e.printStackTrace();

        }
    }

    /**
     * 将信息群发到其他客户端
     *
     * @param message
     * @param socketChannel
     */
    private void sendInfoToOtClients(String message, SocketChannel socketChannel) {

        //获取所有注册到选择器的客户端，并遍历
        for (SelectionKey selectionKey : selector.keys()) {
            //获取通道
            Channel channel = selectionKey.channel();
            //判断通道是否属于SocketChannel，同时不等于发送信息的客户端
            if (channel instanceof SocketChannel && channel != socketChannel) {
                //通道转换
                SocketChannel sc = (SocketChannel) channel;
                writeMsg2Channel(sc, message);
            }
        }
    }

    /**
     * 将信息发给客户端其他客户端
     *
     * @param message
     * @param socketChannel
     */
    private static void writeMsg2Channel(SocketChannel socketChannel, String message) {
        //将信息写入缓冲区
        ByteBuffer byteBuffer = ByteBuffer.wrap(message.getBytes(StandardCharsets.UTF_8));
        try {
            //将缓冲区的数据写入通道
            socketChannel.write(byteBuffer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}